package com.mjf.spark.day06

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 向MySQL数据库中写入数据
 */
object Spark05_MySQL_Write {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark05_MySQL_Write")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    // 数据库连接四要素
    var driver = "com.mysql.jdbc.Driver"
    var url = "jdbc:mysql://hadoop103:3306/test"
    var username = "root"
    var password = "123456"

    /**
     * JDBC操作步骤：
     *    1.注册驱动
     *    2.获取连接
     *    3.创建数据库操作对象PrepareStatement
     *    4.执行SQL
     *    5.处理结果集
     *    6.关闭连接
     */
    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("lucy", 22), ("jack", 24)))

    // 下面这段代码，需要PreparedStatement实现序列化，但是PS不是我们自己定义的类型，没有办法实现
    /*
    // 1.注册驱动
    Class.forName(driver)
    // 2.获取连接
    val conn: Connection = DriverManager.getConnection(url, username, password)
    // 声明数据库操作SQL语句
    var sql: String = "insert into user(name,age) values(?,?)"
    // 3.创建数据库操作对象PrepareStatement
    val ps: PreparedStatement = conn.prepareStatement(sql)

    rdd.foreach{
      case (name, age) => {
        // 给参数赋值
        ps.setString(1, name)
        ps.setInt(2, age)
        // 4.执行SQL
        ps.executeUpdate()
        // 5.处理结果集  不是查询不需要
      }
    }

    // 6.关闭连接
    ps.close()
    conn.close()
    */


    rdd.foreachPartition{
      // datas是RDD的一个分区的数据
      datas => {
        // 1.注册驱动
        Class.forName(driver)
        // 2.获取连接
        val conn: Connection = DriverManager.getConnection(url, username, password)
        // 声明数据库操作SQL语句
        var sql: String = "insert into user(name,age) values(?,?)"
        // 3.创建数据库操作对象PrepareStatement
        val ps: PreparedStatement = conn.prepareStatement(sql)

        // 对当前分区内数据，进行遍历
        // 注意：这个foreach不是算子，是集合的方法
        datas.foreach{
          case (name, age) => {
            // 给参数赋值
            ps.setString(1, name)
            ps.setInt(2, age)
            // 4.执行SQL
            ps.executeUpdate()
          }
        }

        // 6.关闭连接
        ps.close()
        conn.close()
      }
    }

    // 关闭连接
    sc. stop()

  }
}
